package com.leiyuee.flink.tools.enums.config;

import com.alibaba.fastjson.JSONObject;
import org.apache.flink.table.data.GenericRowData;
import org.apache.flink.table.data.RowData;
import org.apache.flink.table.data.StringData;
import org.apache.flink.table.data.TimestampData;

import java.text.SimpleDateFormat;
import java.util.function.Function;


/**
 * JSON 转 RowData Function 配置
 */
public interface ConvertFunction {

    /**
     * IOTLink Kafka 数据 JSON 转 RowData Function
     */
    Function<JSONObject, RowData> IOT_LINKER = json -> {
        GenericRowData rowData = new GenericRowData(15);
        // 设置字段值
        rowData.setField(0, StringData.fromString(json.getString("Id")));
        rowData.setField(1, StringData.fromString(json.getString("Sender")));
        rowData.setField(2, StringData.fromString(json.getString("PanelId")));
        rowData.setField(3, StringData.fromString(json.getString("MsgType")));

        String createTime = json.getString("CreateTime");
        String ts = createTime.substring(0, 23);
        String partitionCreateTimeYear = createTime.substring(0, 4);
        String partitionCreateTimeMon = createTime.substring(5, 7);
        String partitionCreateTimeDay = createTime.substring(8, 10);
        String partitionCreateTimeHour = createTime.substring(11, 13);
        rowData.setField(4, StringData.fromString(createTime));

        rowData.setField(5, StringData.fromString(json.getString("Payload")));
        rowData.setField(6, StringData.fromString(ts));
        rowData.setField(7, StringData.fromString(json.getString("TimeStamp")));
        rowData.setField(8, StringData.fromString(json.getString("OriginalTimeStamp")));
        rowData.setField(9, StringData.fromString(json.getString("OriginalCreatetime")));
        rowData.setField(10, StringData.fromString(json.getString("HostName")));

        rowData.setField(11, StringData.fromString(partitionCreateTimeYear));// partition_CreateTime_year
        rowData.setField(12, StringData.fromString(partitionCreateTimeMon));// partition_CreateTime_mon
        rowData.setField(13, StringData.fromString(partitionCreateTimeDay));// partition_CreateTime_day
        rowData.setField(14, StringData.fromString(partitionCreateTimeHour));// partition_CreateTime_hour
        return rowData;
    };

    Function<JSONObject, RowData> KAFKA_KMRECORD_TO_DORIS_ATTENDANCERECORDRT = json -> {
        GenericRowData rowData = new GenericRowData(10);// 原长度 15
        rowData.setField(0, StringData.fromString(json.getString("CardNo")));// CardNo
        rowData.setField(1, TimestampData.fromEpochMillis(json.getLongValue("FTime")));// Ftime
        rowData.setField(2, StringData.fromString(json.getString("DeviceNo")));// DeviceNo
        rowData.setField(3, (byte) json.getIntValue("FType"));// Ftype
        //rowData.setField(4, null);// Empno
        //rowData.setField(5, null);// Fname
        //rowData.setField(6, null);// Place
        //rowData.setField(7, null);// IP
        //rowData.setField(8, null);// Note
        // 在 Java 中，tinyint 数据类型通常可以对应为 byte 类型
        rowData.setField(4, (byte) 1);// cnt
        //rowData.setField(9, (byte) 1);// cnt

        long timestamp = System.currentTimeMillis();// 当前 UTC 时间毫秒数
        String timeString = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format( json.getLongValue("FTime"));
        String year = timeString.substring(0, 4);
        String mon = timeString.substring(5, 7);
        String day = timeString.substring(8, 10);
        String hour = timeString.substring(11, 13);

        //rowData.setField(10, StringData.fromString(year));// partition_FTime_year
        //rowData.setField(11, StringData.fromString(mon));// partition_FTime_mon
        //rowData.setField(12, StringData.fromString(day));// partition_FTime_day
        //rowData.setField(13, StringData.fromString(hour));// partition_FTime_hour
        //rowData.setField(14, TimestampData.fromEpochMillis(timestamp));// ts -> sync_update_time

        rowData.setField(5, StringData.fromString(year));// partition_FTime_year
        rowData.setField(6, StringData.fromString(mon));// partition_FTime_mon
        rowData.setField(7, StringData.fromString(day));// partition_FTime_day
        rowData.setField(8, StringData.fromString(hour));// partition_FTime_hour
        rowData.setField(9, TimestampData.fromEpochMillis(timestamp));// ts -> sync_update_time
        return rowData;
    };

    /**
     * Kafka 转 Doris 表 HR_security.visitorRegistration
     */
    Function<JSONObject, RowData> KAFKA2DORIS_HR_SECURITY_VISITOR_REGISTRATION = json -> {
        GenericRowData rowData = new GenericRowData(22);
        // t_vms_client_visitor_registration：用于保存访客登记记录，每次的进出工厂记录
        rowData.setField(0, json.getIntValue("ID"));// visitor_reg_id
        rowData.setField(1, json.getIntValue("SecPostId"));// security_post_id
        rowData.setField(2, StringData.fromString(json.getString("RecordNo")));// visit_record_number
        rowData.setField(3, StringData.fromString(json.getString("UserId")));// visitor_sys_id
        rowData.setField(4, StringData.fromString(json.getString("CustomNo")));// client_number
        rowData.setField(5, json.getByteValue("OperateType"));// reg_operate_type
        rowData.setField(6, StringData.fromString(json.getString("Remark")));// reg_remark
        rowData.setField(7, StringData.fromString(json.getString("CreateBy")));// reg_creator
        rowData.setField(8, TimestampData.fromEpochMillis(json.getLongValue("CreateDate")));// reg_create_time
        rowData.setField(9, StringData.fromString(json.getString("LastUpdateby")));// reg_updator
        rowData.setField(10, TimestampData.fromEpochMillis(json.getLongValue("CreateDate")));// reg_update_time
        rowData.setField(11, StringData.fromString(json.getString("Extend01")));// extended_field_one
        rowData.setField(12, StringData.fromString(json.getString("Extend02")));// extended_field_two
        rowData.setField(13, StringData.fromString(json.getString("Security_Card")));// security_badge_id
        rowData.setField(14, StringData.fromString(json.getString("Receptionist_Card")));// receptionist_badge_id

        // 补上缺少的字段 cnt, ts, partition_createdate_year, partition_createdate_mon, partition_createdate_day, partition_createdate_hour, sync_update_time
        rowData.setField(15, 1);// cnt
        long timestamp = System.currentTimeMillis();
        String tsString = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(timestamp);
        String timeString = new SimpleDateFormat("yyyy-MM-dd HH:mm:ss").format(json.getLongValue("CreateDate"));
        String year = timeString.substring(0, 4);
        String mon = timeString.substring(5, 7);
        String day = timeString.substring(8, 10);
        String hour = timeString.substring(11, 13);

        rowData.setField(16, StringData.fromString(tsString));// ts

        rowData.setField(17, StringData.fromString(year));// partition_createdate_year
        rowData.setField(18, StringData.fromString(mon));// partition_createdate_mon
        rowData.setField(19, StringData.fromString(day));// partition_createdate_day
        rowData.setField(20, StringData.fromString(hour));// partition_createdate_hour

        rowData.setField(21, TimestampData.fromEpochMillis(timestamp));// sync_update_time

        return rowData;
    };
}
